package com.penghk.springbootdemo.rabbitMQ;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeoutException;

public class Publisher {


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

            ConnectionFactory factory = new ConnectionFactory();
            factory.useNio();

            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setVirtualHost("/");
            Connection connection = factory.newConnection();

            //默认情况下客户端会自动分配一个ExecutorService给Consumer线程，同样你也可以使用自定义的线程池
            //当connection关闭的时候，默认的ExecutorService会被shutdown,但是如果是自定义的ExecutorService将不会被自动的shutdown，所以Clients程序需要在最终关闭的时候手动的去执行shutdown()，否则将会阻止JVM的正常关闭
            //除非有明显的证据证明默认的ExecutorService不能满足当前Consumer callbacks的需要，否则不建议使用自定义的ExecutorService.
//            ExecutorService ex = Executors.newFixedThreadPool(10);
//            connection = factory.newConnection(ex);


            //另一种方式创建连接
//        factory.setUri("amqp://guest:guest@127.0.0.1:5762//");
//        Connection connection = factory.newConnection();

            Channel channel = connection.createChannel();
            //在使用之前必须先“declared”(声明)，确保在使用之前已经存在，如果不存在则创建它，这些操作都包含在declare里
            String exchangeName = "myDemoExchange1";
            channel.exchangeDeclare(exchangeName, "direct", true);

            //排他的（只对当前client同一个Connection可用, 同一个Connection的不同的Channel可共用），并且也会在client连接断开时自动删除
//            String queueName = channel.queueDeclare().getQueue();
//            channel.queueBind(queueName, exchangeName, "key1");

            String queueName1 = "phkQueue1";
            channel.queueDeclare(queueName1, true, false,false, null);
            String routingKey = "key1";
            channel.queueBind(queueName1, exchangeName, routingKey);

            byte[] messages = "hello world".getBytes();
//            channel.basicPublish(exchangeName, routingKey, null, messages);

            //delivery mode为2，即消息需要被持久化在broker中
            //如果一个消息在publish的时候设置了mandatory标记，如果消息没有成功的路由到某个队列的时候，broker端会通过Basic.Return返回回来。
            //
            //这时候客户端需要实现ReturnListener这个接口，并且调用Channel.setReturnListener
            //消息通过Channel,basicPublish发布，如果connection down了那么消息就会丢失
            channel.basicPublish(exchangeName, routingKey, true, MessageProperties.PERSISTENT_TEXT_PLAIN, messages);
//            System.out.println("publish success");



            SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() {
                    @Override
                    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                            System.out.println("ack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                            if (multiple) {
                                    confirmSet.headSet(deliveryTag+1).clear();
                            } else {
                                    confirmSet.remove(deliveryTag);
                            }
                    }

                    @Override
                    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                            System.out.println("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
                            if (multiple) {
                                    confirmSet.headSet(deliveryTag + 1).clear();
                            } else {
                                    confirmSet.remove(deliveryTag);
                            }
                    }
            });

//            for (int i = 0;i < 100000; i++){
            for (int i = 0;i < 10; i++){
                    long nextSeqNo = channel.getNextPublishSeqNo();

                    Map<String, Object> headers = new HashMap<String, Object>();
                    headers.put("latitude",  51.5252949);
                    headers.put("longitude", -0.0905493);
                    channel.basicPublish(exchangeName, routingKey, new AMQP.BasicProperties.Builder()
                            .headers(headers)
                            .expiration("10000")//存活时间　10s
                            .build(), messages);
                    confirmSet.add(nextSeqNo);
//                    Thread.sleep(100);
            }


            connection.close();
    }
}
